package cn.doitedu.rules.dao;

import cn.doitedu.date.gen.LogBean;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;
import java.util.Set;
   /*将hbase具体的查询语句和数据库操作写入到固定的程序中*/
public class UserProfileHbaseDaoImpl implements UserProfileDao {

    private Connection hbaseConn;
    //通过构造方法获取到制定的构造器
    public UserProfileHbaseDaoImpl(Connection hbaseConn) {
        this.hbaseConn = hbaseConn;
    }

    @Override
    public boolean compareTags(LogBean event, Map<String, String> userProfiles) throws Exception {

        if (userProfiles != null) {
            Table table = hbaseConn.getTable(TableName.valueOf("zenniu_profile"));
            //获取当前输入的事件的设备ID，作为查询hbase的row-key
            Get get = new Get(Bytes.toBytes(event.getDeviceId()));
            Set<String> userProfileKeys = userProfiles.keySet();
            //指定查询的列族和列的标识符
            for (String profileKey : userProfileKeys) {
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(profileKey));
            }
            Result result = table.get(get);
            table.close();
            if (result != null) {
                for (String profileKey : userProfileKeys) {
                    //预设的画像标签值
                    String targetValue = userProfiles.get(profileKey);
                    //实际查出来的标签值
                    String tagValue = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes(profileKey)));
                    if (!targetValue.equals(tagValue)) {
                        return false;
                    }
                }
                return true;
            } else {
                return false;
            }
        } else {
            return true;
        }
    }
}
